package com.study.flink.java.day07_join_count;

import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import com.alibaba.fastjson.JSON;
import com.study.flink.java.day07_join_count.entity.OrderDetail;
import com.study.flink.java.day07_join_count.entity.OrderMain;
import com.study.flink.java.utils.FlinkUtils;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * 订单明细表和订单主表进行join
 * 使用EvenTime划分滚动窗口
 * 使用CoGroup实时左外连接
 * 没有Join上的数据单独处理
 * 还没试验，有空再弄
 */
public class OrderJoinMain {

    private static final Log log = LogFactory.get();

    public static void main(String[] args) throws Exception{

        ParameterTool parameters = ParameterTool.fromPropertiesFile(args[0]);

        long windowTime = parameters.getLong("window.time", 2000);
        long delayTime = parameters.getLong("delay.time", 2000);

        StreamExecutionEnvironment env = FlinkUtils.getEnv();
        env.setParallelism(1); // 测试用

        //设置使用EventTime作为时间标准
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<String> OrderDetailStream = FlinkUtils.createKafkaStream(parameters,
                parameters.getRequired("order.detail.topics"),
                parameters.getRequired("order.detail.gid"),
                SimpleStringSchema.class);

        //底层方法实现FlatMap和Fliter方法
        SingleOutputStreamOperator<OrderDetail> orderDetailBeanStream = OrderDetailStream.process(new ProcessFunction<String, OrderDetail>() {
            @Override
            public void processElement(String value, Context context, Collector<OrderDetail> out) throws Exception {
                try{
                    OrderDetail orderDetail = JSON.parseObject(value, OrderDetail.class);
                    out.collect(orderDetail);
                } catch (Exception e) {
                    log.error(e);
                }
            }
        });

        //提取EventTime并生产WaterMark
        SingleOutputStreamOperator<OrderDetail> orderDetailWithWaterMark = orderDetailBeanStream.assignTimestampsAndWatermarks(
                new BoundedOutOfOrdernessTimestampExtractor<OrderDetail>(Time.milliseconds(delayTime)) {
            @Override
            public long extractTimestamp(OrderDetail orderDetail) {
                return orderDetail.updateTime.getTime();
            }
        });

        DataStream<String> OrderMainStream = FlinkUtils.createKafkaStream(parameters,
                parameters.getRequired("order.main.topics"),
                parameters.getRequired("order.main.gid"),
                SimpleStringSchema.class);
        SingleOutputStreamOperator<OrderMain> orderMainBeanStream = OrderMainStream.process(new ProcessFunction<String, OrderMain>() {
            @Override
            public void processElement(String value, Context context, Collector<OrderMain> out) throws Exception {
                try{
                    OrderMain OrderMain = JSON.parseObject(value, OrderMain.class);
                    out.collect(OrderMain);
                } catch (Exception e) {
                    log.error(e);
                }
            }
        });

        SingleOutputStreamOperator<OrderMain> orderMainWithWaterMark = orderMainBeanStream.assignTimestampsAndWatermarks(
                new BoundedOutOfOrdernessTimestampExtractor<OrderMain>(Time.milliseconds(delayTime)) {
            @Override
            public long extractTimestamp(OrderMain orderMain) {
                return orderMain.updateTime.getTime();
            }
        });

        // 明细表作为左表(作为大宽表写入hdfs)
        // 左表数据Join右表数据，右表数据迟到了，没有join上，再查库关联右表数据
        // 左表数据迟到：使用侧流输出获取迟到的数据
        DataStream<Tuple2<OrderDetail, OrderMain>> results = orderDetailWithWaterMark.coGroup(orderMainWithWaterMark)
                .where(new KeySelector<OrderDetail, Long>() {
                    @Override
                    public Long getKey(OrderDetail orderDetail) throws Exception {
                        return orderDetail.orderId;
                    }
                })
                .equalTo(new KeySelector<OrderMain, Long>() {
                    @Override
                    public Long getKey(OrderMain orderMain) throws Exception {
                        return orderMain.oid;
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.milliseconds(windowTime)))
                .apply(new CoGroupFunction<OrderDetail, OrderMain, Tuple2<OrderDetail, OrderMain>>() {
                    @Override
                    public void coGroup(Iterable<OrderDetail> OrderDetails, Iterable<OrderMain> OrderMains, Collector<Tuple2<OrderDetail, OrderMain>> out) throws Exception {
                        //先选好左表的数据
                        for (OrderDetail orderDetail : OrderDetails) {
                            boolean isJoined = false;
                            for (OrderMain orderMain : OrderMains) {
                                out.collect(Tuple2.of(orderDetail, orderMain));
                                isJoined = true;
                            }
                            if (!isJoined) {
                                out.collect(Tuple2.of(orderDetail, null));
                            }
                        }
                    }
                });

        results.print();

        FlinkUtils.getEnv().execute("OrderJoinMain");
    }

}
